package com.jushu.storbox.task.tqueue;

import java.util.Collection;
import java.util.Comparator;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.PriorityBlockingQueue;

public abstract class TaskScheduler <T extends PriorityTask>{
	
	private final ExecutorService service;
	private final PriorityBlockingQueue<T> taskQueue;
	private volatile boolean schedulerStoped = true;
	
	public TaskScheduler(int initialCapacity) {
		taskQueue = new PriorityBlockingQueue<T>(initialCapacity, new Comparator<T>() {

			@Override
			public int compare(T task1, T task2) {
				if(null == task1 && null == task2) {
					return 0;
				}
				if (null == task1 || null == task2) {
					if (null == task1) {
						return -1;
					} else {
						return 1;
					}
				}
				if (task1.hashCode() == task2.hashCode()) {
					return 0;
				}
				
				if(task2.getPriority() != task1.getPriority()) {
					return task2.getPriority() - task1.getPriority();
				} else {
					return (int)(task2.getSequenceNumber() - task1.getSequenceNumber());
				}
			}
			
		});
		service = Executors.newSingleThreadExecutor();
	}
	
	public void addTaskToScheduler(T task) {
		synchronized(taskQueue) {
			taskQueue.add(task);
			taskQueue.notifyAll();
		}
	}
	public void addTasksToScheduler(Collection <T> taskCollection) {
		synchronized(taskQueue) {
			taskQueue.addAll(taskCollection);
			taskQueue.notifyAll();
		}
	}
	public boolean removeTask(T task) {
		synchronized(taskQueue) {
			return taskQueue.remove(task);
		}
	}
	public boolean contains(T task){
		synchronized(taskQueue) {
			return taskQueue.contains(task);
		}
	}
	public Set<T> getTaskQueue() {
		synchronized(taskQueue) {
			return new HashSet<T>(taskQueue);
		}
	}
	public int size() {
		synchronized(taskQueue) {
			return taskQueue.size();
		}
	}

	public abstract void stopTaskExecuteService();
	public abstract void startTaskExecuteService();
	protected abstract void executeTask(T task);
	
	public synchronized void startScheduler() {
		if(schedulerStoped) {
			schedulerStoped = false;
			service.execute(new Runnable() {
				@Override
				public void run() {
					while(!schedulerStoped) {
						try {
							synchronized(taskQueue) {
								taskQueue.wait();
							}
							
							if(schedulerStoped) {
								break;
							}
							while(!taskQueue.isEmpty() && !schedulerStoped && !Thread.interrupted()) {
								T task = taskQueue.poll();
								try {
									executeTask(task);
								} catch(Exception ex) {
									ex.printStackTrace();
								}
							}
						} catch(Exception e) {
						}
					}
				}
			});
		}
	}
	public void stopScheduler() {
		if(schedulerStoped) {
			return;
		}
		schedulerStoped = true;
		synchronized(taskQueue) {
			taskQueue.notifyAll();
		}
		stopTaskExecuteService();
	}
	
	protected class Lock {
	}
}
